package com.xxxx;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;


public class Hello03WaterMark {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        //水印使用前提，基于（事件时间）时间窗口
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<String> source = env.socketTextStream("192.168.88.101", 18880);

        SingleOutputStreamOperator<String> andWatermarks = source.assignTimestampsAndWatermarks(new PunctuatedWaterMark());

        andWatermarks.map(word -> Tuple2.of(word, 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(0)
                .timeWindow(Time.seconds(5))
                .sum(1)
                .print();
        env.execute();
    }
}

class PunctuatedWaterMark implements AssignerWithPunctuatedWatermarks<String> {

    @Override
    public Watermark checkAndGetNextWatermark(String line, long l) {
        if (line != null && "hello".equals(line)) {
            return new Watermark(System.currentTimeMillis());
        } else {
            return null;
        }

    }

    @Override
    public long extractTimestamp(String line, long previousElementTimestamp) {
        long timestamp = System.currentTimeMillis();
        System.out.println("[" + line + "][" + timestamp + "]");
        return timestamp;
    }
}